// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once

#include <algorithm>
#include <cstdint>
#include <deque>
#include <limits>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include <boost/crc.hpp>

#include "coin2/base/concurrent_queue.h"
#include "coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/exchange/base/executor_util/stat.h"

namespace coin2::exchange::base::executor {

class TopicRecordProcessor {
  // cannot public TopicRecordSubscriber due to const-less
 public:
  explicit TopicRecordProcessor(TopicRecordSubscriber* subscriber) : subscriber_(subscriber) {}

  void OnTopicRecord(TopicRecordPacket* packet) {
    bool exception = false;
    bool keyboard_interrupt = false;
    packet->distributor_rx_timestamp = GetTimestamp();
    subscriber_->OnTopicRecordBase(*packet);
    /*
    try {
      subscriber_->OnTopicRecord(packet);
    } catch (const std::exception& e) {
      exception = true;
      LOG(ERROR) << "Exception on message processing (1/4): " << packet.topic_name;
      LOG(ERROR) << "Exception on message processing (2/4): " << e.what();
      LOG(ERROR) << "Exception on message processing (3/4): " << packet.sub_topic_id;
      LOG(ERROR) << "Exception on message processing (4/4): " << packet.payload;

      // python's KeyboardInterrupt
      if (strstr(e.what(), "KeyboardInterrupt") != nullptr) {
        keyboard_interrupt = true;
        LOG(ERROR) << "KeyboardInterrupt";
      } else {
        throw;
      }
    }*/
    packet->distributor_tx_timestamp = GetTimestamp();
    stat_.Add(packet->distributor_rx_timestamp, packet->distributor_tx_timestamp, exception);
    (void)keyboard_interrupt;
  }
  void PrintDebugMessage(const std::string& topic_name) const {
    stat_.PrintDebugMessage(topic_name);
  }

  static int64_t GetTimestamp() {
    auto now = std::chrono::high_resolution_clock::now();
    return now.time_since_epoch().count();
  }

 protected:
  TopicRecordSubscriber* subscriber_;
  ProcessorStat stat_;
};

class TopicRecordArbitrator : public TopicRecordSubscriber {
 public:
  explicit TopicRecordArbitrator(TopicRecordSubscriber* subscriber, int64_t window_size)
      : subscriber_(subscriber), window_size_(window_size) {
    CHECK_GE(crc_deque_.max_size(), window_size);
    CHECK_GE(crc_set_.max_size(), window_size);
  }

  void OnTopicRecord(const TopicRecordPacket& packet) override {
    boost::crc_32_type result;
    result.process_bytes(packet.payload.c_str(), packet.payload.size());
    int32_t crc = result.checksum();
    if (crc != 0 && crc_set_.find(crc) != crc_set_.end()) {
      // already exists. do not publish.
    } else {
      subscriber_->OnTopicRecordBase(packet);
      crc_set_.insert(crc);
    }
    crc_deque_.push_back(crc);
    // THINK: how many?
    if (crc_deque_.size() > window_size_) {
      crc = crc_deque_.front();
      crc_set_.erase(crc);
      crc_deque_.pop_front();
    }
  }

 protected:
  TopicRecordSubscriber* subscriber_;
  std::deque<int32_t> crc_deque_;
  std::set<int32_t> crc_set_;
  int64_t window_size_;
};

class TopicRecordDistributor {
 public:  // user
  std::vector<std::string> GetTopics() const {
    std::vector<std::string> keys;
    std::transform(
        subscriber_map_.begin(),
        subscriber_map_.end(),
        std::back_inserter(keys),
        [](const auto& pair) { return pair.first; });
    return keys;
  }

  void LogTopic() const {
    VLOG(1) << "registered topics: ";
    for (const auto& it : subscriber_map_) {
      VLOG(1) << "  " << it.first << " " << it.second.size();
    }
  }

  void Subscribe(const std::string& topic, TopicRecordSubscriber* subscriber) {
    CHECK(subscriber);
    subscriber_map_[topic].push_back(TopicRecordProcessor(subscriber));
  }

  void Publish(const std::string& topic) {
    const auto& vec = subscriber_map_[topic];
    if (vec.empty()) {
      LOG(WARNING) << "no one subscribes topic: " << topic;
    }
  }

  virtual void OnTopicRecord(const std::string& topic, TopicRecordPacket* packet) {
    auto& vec = subscriber_map_[topic];

    if (vec.empty()) {
      LOG(WARNING) << "no one subscribes topic: " << topic;
    }

    CHECK(!(vec.empty())) << topic;  // no callback fo this topic
    for (auto& out : vec) {
      out.OnTopicRecord(packet);
      // if (false) {  // print_stat
      //   out.PrintDebugMessage(topic);
      // }
    }
  }

  virtual void Stop() {}

 protected:
  std::map<std::string, std::vector<TopicRecordProcessor>> subscriber_map_;
};

class TopicRecordDistributorMultiThread : public TopicRecordDistributor{
 public:
  explicit TopicRecordDistributorMultiThread(int thread_num)
      : stop_(false) {
    for (int i = 0; i < thread_num; ++i) {
      auto queue = std::make_unique<ConcurrentQueue<std::shared_ptr<TopicRecordPacket>>>(500);
      auto thread = std::make_unique<std::thread>(std::bind(
          &TopicRecordDistributorMultiThread::RunOnTopicRecord, this, i, queue.get()));
      queues_.push_back(std::move(queue));
      threads_.push_back(std::move(thread));
    }
  }
  void OnTopicRecord(const std::string& topic, TopicRecordPacket* packet) override {
    auto& vec = subscriber_map_[topic];

    if (vec.empty()) {
      LOG(WARNING) << "no one subscribes topic: " << topic;
    }

    CHECK(!(vec.empty())) << topic;  // no callback fo this topic
    for (int i = 0; i < vec.size(); ++i) {
      auto buf = std::make_shared<exchange::base::executor::TopicRecordPacket>();
      *buf = *packet;
      queues_[i]->Push(buf);
    }
  }
  void Stop() override {
    stop_ = true;
    for (const auto& queue : queues_) {
      queue->Push(nullptr);
      queue->NotifyAll();
    }
    for (auto& thread : threads_) {
      thread->join();
    }
  }

 private:
  void RunOnTopicRecord(int idx, ConcurrentQueue<std::shared_ptr<TopicRecordPacket>>* queue) {
    while (true) {
      if (queue->IsEmpty() && stop_) {
        break;
      }
      auto packet = queue->Pop();
      if (packet == nullptr) break;
      subscriber_map_[packet->topic_name][idx].OnTopicRecord(packet.get());
    }
  }
  std::vector<std::unique_ptr<ConcurrentQueue<std::shared_ptr<TopicRecordPacket>>>> queues_;
  std::vector<std::unique_ptr<std::thread>> threads_;
  bool stop_;
};

}  // namespace coin2::exchange::base::executor
